package com.arch.flink.cep;

import com.arch.flink.cep.bean.PayEvent;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * https://cloud.tencent.com/developer/article/1779977
 * <p>
 * 找出那些下单后 10 分钟内没有支付的订单
 *
 * @author pizhihui
 * @date 2023-10-24 14:34
 */
public class TimeoutPayCEPMain {


    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<PayEvent> source = env.fromElements(new PayEvent(1L, "create", 1597905234000L),
                new PayEvent(1L, "pay", 1597905235000L),
                new PayEvent(2L, "create", 1597905236000L),
                new PayEvent(2L, "pay", 1597905237000L),
                new PayEvent(3L, "create", 1597905239000L));


        source.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<PayEvent>(Time.milliseconds(500L)) {
            @Override
            public long extractTimestamp(PayEvent element) {
                return element.getTimestamp();
            }
        }).keyBy(new KeySelector<PayEvent, Object>() {
            @Override
            public Object getKey(PayEvent value) throws Exception {
                return value.getId();
            }
        });

        // 逻辑处理带阿米
        OutputTag<PayEvent> orderTimeoutOutputTag = new OutputTag<PayEvent>("orderTimeout") {
        };

        Pattern<PayEvent, PayEvent> pattern = Pattern.<PayEvent>
                        begin("begin")
                .where(new IterativeCondition<PayEvent>() {
                    @Override
                    public boolean filter(PayEvent payEvent, Context<PayEvent> context) throws Exception {
                        return payEvent.getOp().equals("create");
                    }
                })
                .followedBy("pay")
                .where(new IterativeCondition<PayEvent>() {
                    @Override
                    public boolean filter(PayEvent payEvent, Context<PayEvent> context) throws Exception {
                        return payEvent.getOp().equals("pay");
                    }
                })
                .within(Time.minutes(10));

        PatternStream<PayEvent> patternStream = CEP.pattern(source, pattern);
        SingleOutputStreamOperator<Object> result = patternStream.select(orderTimeoutOutputTag, new PatternTimeoutFunction<PayEvent, PayEvent>() {
            @Override
            public PayEvent timeout(Map<String, List<PayEvent>> map, long l) throws Exception {
                return map.get("begin").get(0);
            }
        }, new PatternSelectFunction<PayEvent, Object>() {
            @Override
            public Object select(Map<String, List<PayEvent>> map) throws Exception {
                return map.get("pay").get(0);
            }
        });

//        result.print();
//
        SideOutputDataStream<PayEvent> sideOutput = result.getSideOutput(orderTimeoutOutputTag);
        sideOutput.print();


        env.execute("execute timeout pay");

    }

}
